package com.anyang.mq.consumer;

import com.anyang.mq.common.domain.Rocketmq;
import com.anyang.mq.common.service.RocketmqService;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.time.LocalDateTime;

/**
 * Created by anyang
 * Email: 1300378587@qq.com
 * Created Date:2019-11-12 09:48
 */

@Slf4j
@Component
@RocketMQMessageListener(topic = "xst-mq",consumerGroup = "${rocketmq.consumer.group}",messageModel = MessageModel.CLUSTERING)
//订阅主题= "springboot-rocketmq"，消费者组， messageModel：消费模式【广播，负载均衡】
public class Cusumer implements RocketMQListener<MessageExt>{
    @Autowired
    RocketmqService rocketmqService;
    @Override
    public void onMessage(MessageExt message){
        String msgId = message.getMsgId();
        String body = new String(message.getBody());
        log.info("收到消息:: "+msgId);
        //创建对象保存进程的情况
        //sdsdsd
        Rocketmq rocketmq = new Rocketmq();
        rocketmq.setInfo("massage received " + msgId);
        rocketmq.setResult(body);
        rocketmq.setType("1");
        boolean save = rocketmqService.save(rocketmq);
        log.info("保存消息:: "+message+" "+(save?"成功":"失败")+ LocalDateTime.now());
        
    }
}

